package com.star.sync.elasticsearch.listener;

import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.protobuf.InvalidProtocolBufferException;
import com.star.sync.elasticsearch.dao.BaseDao;
import com.star.sync.elasticsearch.event.CanalEvent;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.jdbc.BadSqlGrammarException;
import org.springframework.jdbc.UncategorizedSQLException;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * @program: canal-mysql-ddl-sync
 * @description:
 * @author: young
 * @create: 2018-07-20 16:47
 **/
@Component
public class DDLCanalListener<DDLCanalEvent extends CanalEvent> implements ApplicationListener<DDLCanalEvent> {

    private static final Logger logger = LoggerFactory.getLogger(DDLCanalListener.class);
    @Resource
    private BaseDao baseDao;

    @Override
    public void onApplicationEvent(DDLCanalEvent event) {
        CanalEntry.Entry entry = event.getEntry();
        CanalEntry.Header header = entry.getHeader();
        String database = entry.getHeader().getSchemaName();

        if (database.endsWith("_dev") || database.endsWith("_test")) {
            logger.info(database+",开发库不同步！");
            return;
        } else if (database.equals("sys")) {
            logger.info("sys,库不同步！");
            return;
        }

        String table = entry.getHeader().getTableName();
        CanalEntry.EventType eventType = header.getEventType();
        if (CanalEntry.EventType.INSERT == eventType ||
                CanalEntry.EventType.UPDATE == eventType ||
                CanalEntry.EventType.DELETE == eventType ||
                CanalEntry.EventType.TRUNCATE == eventType ||
                CanalEntry.EventType.QUERY == eventType) {
            logger.info("no ddl!");
            return;
        }

        CanalEntry.RowChange change;
        try {
            change = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
        } catch (InvalidProtocolBufferException e) {
            logger.error("canalEntry_parser_error,根据CanalEntry获取RowChange失败！", e);
            return;
        }
        String sql = change.getSql();
        String t = "`"+table+"`";
        if (sql.contains("."+t) || sql.contains("."+table)) {
        } else if (sql.contains(" "+t)) {
            sql = sql.replace("`"+table+"`", database+".`"+table+"`");
        } else if (sql.contains(" "+table)) {
            sql = sql.replace(" "+table, " "+database+"."+table);
        }
        if (CanalEntry.EventType.RENAME == eventType) {
            String sqlTemp = sql.toLowerCase();
            String srcTable = StringUtils.substringBetween(sqlTemp, " table ", " to ");
            if (!srcTable.contains("."))
                sql = sql.replace(" "+srcTable, " "+database+"."+srcTable);
        }
        try {
            baseDao.ddl(sql);
        } catch (BadSqlGrammarException e) {
            logger.warn("BadSqlGrammarException : {}", e);
            String msg = e.getMessage();
            if (StringUtils.isNotBlank(msg) && msg.contains("already exists")) {
                sql = "select 1";
//                baseDao.ddl(sql);
            }
        } catch (UncategorizedSQLException e) {
            logger.warn("UncategorizedSQLException : {}", e);
        }
        logger.info("ddl：{}", sql);
    }
}
